package com.electron.power.reduce;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONObject;
import com.electron.power.model.mod.ModBodyMap;
import com.electron.power.model.mod.ModMapData;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class SocketTextStream {
	public static void main(String[] args) throws Exception {
		// 获取环境
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//		env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
		DataStreamSource<String> stream = env.socketTextStream("localhost", 9000);
		// 计数
		SingleOutputStreamOperator<ModBodyMap> reduce = stream.map(new StringExcangeModBodyMapFunction()).keyBy("deviceId")
				.timeWindow(Time.minutes(1)).reduce(new ReduceFunction<ModBodyMap>() {
					private static final long serialVersionUID = 1L;

					@Override
					public ModBodyMap reduce(ModBodyMap value1, ModBodyMap value2) throws Exception {
						ModBodyMap result = new ModBodyMap();
						ModMapData modMapDataOfA = new ModMapData();
						ModMapData modMapDataOfValue1 = value1.getModMapDataOfA();
						ModMapData modMapDataOfValue2 = value2.getModMapDataOfA();
						modMapDataOfA.setCurrent(
								(modMapDataOfValue1.getCurrent() + modMapDataOfValue2.getCurrent()) / 2);
						modMapDataOfA.setPower(modMapDataOfValue1.getPower()+modMapDataOfValue2.getPower());
						result.setModMapDataOfA(modMapDataOfA);
						return result;
					}
				});

		reduce.print();

		env.execute("Java WordCount from SocketTextStream Example");
//		DataStream<ModBodyMap> reduce =env.socketTextStream("localhost", 9000)
//		        .map(new StringExcangeModBodyMapFunction())
//		        .keyBy(1).timeWindow(Time.seconds(10)).reduce(new ReduceFunction<ModBodyMap>() {
//					
//					@Override
//					public ModBodyMap reduce(ModBodyMap value1, ModBodyMap value2) throws Exception {
//						// TODO Auto-generated method stub
//						return null;
//					}
//				});
//		reduce.print();
//		counts.writeAsCsv("/mnt/flink/output.csv", "\n", " ");

	}

	public static final class LineSplitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
		private static final long serialVersionUID = 1L;

		@Override
		public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) {
			String[] tokens = s.toLowerCase().split("\\W+");

			for (String token : tokens) {
				if (token.length() > 0) {
					collector.collect(new Tuple2<String, Integer>(token, 1));
				}
			}
		}
	}

	public static class StringExcangeModBodyMapFunction implements MapFunction<String, ModBodyMap> {

		private static final long serialVersionUID = -1L;

		@Override
		public ModBodyMap map(String value) throws Exception {
//			ObjectMapper mapper = new ObjectMapper();
			ModBodyMap modBodyMap = JSONObject.parseObject(value, ModBodyMap.class);
			log.info("收取到的数据,{}", modBodyMap);
			return modBodyMap;
		}
	}
}
